package com.globant.AutomateEMR;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import com.globant.Exceptions.BigDataFunctionException;

public class AutomateBigDataFunctions {
	PropertyLoader prop;
	private String jobID;
	
	public String getJobID() {
		return jobID;
	}
	public void setJobID(String jobID) {
		this.jobID = jobID;
	}
	public AutomateBigDataFunctions(PropertyLoader prop) {
		super();
		this.prop = prop;
	}
	
//======================================================================================================================================
//automateBigDataFunctions Method
//======================================================================================================================================

	public boolean automateBigDataFunctions(File file){
		String state=null;
		boolean processComplete=false;
		try{
			String jobId=this.runBigDataScrip(file);
			state=this.waitBeforeFinishProcessing(jobId);
			if("COMPLETED".equals(state)){
			//	moveLog(file);
				processComplete=true;
			}
		}catch(Exception e){
			e.printStackTrace();
		}
		return processComplete;
	}
//======================================================================================================================================
//moveLog Method
//======================================================================================================================================

	private void moveLog(File file) throws BigDataFunctionException {
		FileManager fmanager=new FileManager();
		String path2InputProcessedLogs;
		try {
			path2InputProcessedLogs = prop.readProperty("PathToBucket")+prop.readProperty("PathToInputProcessedLogs");
		} catch (IOException e) {
			e.printStackTrace();
			throw new BigDataFunctionException("There was a problem while running the BigData scripts (1)");
		}
		System.out.println("Moving input logs..");
		fmanager.moveFile(file,path2InputProcessedLogs);
		System.out.println("Moved!"+file+" to: "+path2InputProcessedLogs);
	}
	
//======================================================================================================================================
//runBigDataScript Methods
//======================================================================================================================================

	
	public String runBigDataScrip(File file) throws BigDataFunctionException {
		String jobID = "";
		try {
			System.out.println("Running Big Data Script..");
			//String file_name=Avoids3fsError(file.getName());
			String pathToS3File = prop.readProperty("PathToInputLogs")
					+ file.getName();
			Process p = Runtime.getRuntime().exec(prop.readProperty("PathToScripts") + "scriptBigData.sh "
							+ pathToS3File + " "+ prop.readProperty("AbsPathToElasticMapReduce"));
			
			jobID = extractJobId(p);
		} catch (IOException e) {
			// LOG DATABASE!!! This is handled at the highest level, in the automateBDProcessing(), AutomateBigDataProcessing class
			e.printStackTrace();
			throw new BigDataFunctionException("There was a problem while running the BigData scripts (2)");
		}
		return jobID;
	}

	/* ------------------------------------------------------------------------------------------ */

	public String extractJobId(Process p) throws IOException {
		String jobId="";
		BufferedReader stdInput = new BufferedReader(new InputStreamReader(
				p.getInputStream()));
	
		BufferedReader stdError = new BufferedReader(new InputStreamReader(
				p.getErrorStream()));
		jobId=this.getJobId(stdInput,stdError);
		return jobId;
	}
	/* ------------------------------------------------------------------------------------------ */

	public String getJobId(BufferedReader stdInput,BufferedReader stdError) throws IOException{
		String s=null;
		String jobID=null;
		System.out.println("Here is the standard output of the command:\n");
		while ((s = stdInput.readLine()) != null) {
			System.out.println(s);
			if (s.contains("Job Id:")) {
				jobID = s.substring(s.indexOf("Job Id:") + 7);
			}
		}

		// read any errors from the attempted command
		System.out.println("Here is the standard error of the command (if any):\n");
		while ((s = stdError.readLine()) != null) {
			System.out.println(s);
		}
		return jobID;
	}


//======================================================================================================================================
// WaitBeforeFinishingProcessing Methods
//======================================================================================================================================


	public String waitBeforeFinishProcessing(String jobId) throws BigDataFunctionException{
		String state = null;
		try{
			System.out.println("Waiting  before the EMR finishes..");
			Process p = Runtime.getRuntime().exec(prop.readProperty("AbsPathToElasticMapReduce")
												+ "elastic-mapreduce --list " + jobId);
		
			state = getProcessState(jobId, p);
			System.out.println(state);
			while (!"COMPLETED".equals(state) && !"FAILED".equals(state)&& !"TERMINATED".equals(state)) {
				String secondsToWait = prop.readProperty("MinutesBetweenEmrChecking");
				int milis = Integer.parseInt(secondsToWait) * 1000 * 60;
				Thread.sleep(milis);
				p = Runtime.getRuntime().exec(
						prop.readProperty("AbsPathToElasticMapReduce")
								+ "elastic-mapreduce --list " + jobId);
		
				state = getProcessState(jobId, p);
				System.out.println(state);
			}
		}catch(Exception e){
			e.printStackTrace();
			throw new BigDataFunctionException();
		}
		return state;
	}
	/* ------------------------------------------------------------------------------------------ */

	private String getProcessState(String jobId, Process p) throws IOException {
		String state = null;
		BufferedReader stdInput = new BufferedReader(new InputStreamReader(
				p.getInputStream()));

		BufferedReader stdError = new BufferedReader(new InputStreamReader(
				p.getErrorStream()));

		state=this.getState(jobId, stdInput,stdError);
		//System.out.println("OutputScreen: " + state);
		return state;
	}
	/* ------------------------------------------------------------------------------------------ */

	public String getState(String jobId, BufferedReader stdInput,BufferedReader stdError) throws IOException {
		String outputScreen, state = null;

		// read the output from the command
		System.out.println("Here is the standard output of the command:\n");
		while ((outputScreen = stdInput.readLine()) != null) {
			if (outputScreen.contains(jobId)) {
			state = getProcessStateFromScript(jobId, outputScreen);
			}
		}

		// read any errors from the attempted command
		System.out
				.println("Here is the standard error of the command (if any):\n");
		while ((outputScreen = stdError.readLine()) != null) {
			System.out.println(outputScreen);
		}
		//System.out.println("OutputScreen: " + state);
		return state;
	}
	/* ------------------------------------------------------------------------------------------ */

	public String getProcessStateFromScript(String jobId, String outputScreen) {
			outputScreen=this.eliminateSpaces(outputScreen);
			outputScreen = outputScreen.substring(outputScreen.indexOf(jobId)+ jobId.length());
			if(outputScreen.contains("ec2")){
				outputScreen = outputScreen.substring(0,outputScreen.indexOf("ec2-"));
			}else{
				outputScreen = outputScreen.substring(0,outputScreen.indexOf("LogProcessing"));
			}
		return outputScreen;
	}
	/* ------------------------------------------------------------------------------------------ */
	
	private String eliminateSpaces(String outputScreen) {
		Pattern tabPattern = Pattern.compile(" ");	//Eliminate tabs
		Matcher tabMatcher = tabPattern.matcher(outputScreen);
		if(tabMatcher.find()){
			outputScreen=outputScreen.replaceAll(" ", "");
		}
		return outputScreen;
	}
	
	//======================================================================================================================================

	public static void main(String[] args) throws BigDataFunctionException {
		PropertyLoader prop=new PropertyLoader(args[0]);
		AutomateBigDataFunctions autBD=new AutomateBigDataFunctions(prop);
		autBD.waitBeforeFinishProcessing("j-I9IJACTHTJHS");
	}
}
